// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jhkim

#include "coin2/exchange/binance_base/feed_v1/processor.h"

#include <glog/logging.h>
#include <fmt/format.h>

#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/base/symbology/util.h"
#include "coin2/exchange/binance/symbology/product.h"
#include "coin2/exchange/binance_base/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::binance_base::feed_v1::impl {

template<typename TradingDataMutable, typename Parser, typename Publisher>
void ParseTradingData(
    simdjson::ondemand::document& obj,
    std::string_view url,
    int64_t ts,
    TradingDataMutable trading_data,
    Parser parse,
    Publisher publish) {
  auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
  std::string native_symbol = parameters["symbol"];
  // native symbol always is lower
  StringToLowerInPlace(&native_symbol);
  std::string trading_data_interval = parameters["period"];
  auto interval = BinanceBaseFeedUpdateParser::ParseTradingDataInterval(trading_data_interval);
  auto* ratio = trading_data(native_symbol, interval);
  if (ratio != nullptr) {
    simdjson::ondemand::array data = obj["data"].get_array();
    std::optional<int64_t> first_timestamp;
    if (data.count_elements() > 0) {
      first_timestamp = data.at_pointer("/0/timestamp").get<int64_t>() * 1'000'000LL;
      data.reset();
    }

    if (ratio->Size() == 0) {
      ratio->SnapshotClear();
    } else if (first_timestamp &&
                first_timestamp < ratio->RBegin()->timestamp) {
      ratio->SnapshotClear();
    } else {
      ratio->UpdateClear();
    }
    bool to_publish = false;
    for (simdjson::ondemand::object msg : data) {
      bool freezed = parse(msg, ratio);
      to_publish = to_publish || freezed;
    }
    if (to_publish) {
      ratio->SetTimestamp(ts);
      publish();
    }
  }
}

void BinanceBaseFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  if (packet.payload.empty()) return;

  simdjson::padded_string buffer(packet.payload);

  upd_mgr_.onBeginPacket(packet);

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(packet.payload);
  }

  simdjson::ondemand::document obj;

  if (auto err = parser_.iterate(buffer).get(obj); err) {
    if (parser_config_.continue_on_error()) {
      LOG(ERROR) << "[BinanceCommonProcessor] unknown raw msg: " << packet.payload;
    } else {
      throw std::runtime_error(std::string("JSON parse error: ") + simdjson::error_message(err));
    }
    return;
  }

  static constexpr std::string_view kline_url("klines?");
  static constexpr std::string_view wallet_url("capital/config/getall");
  static constexpr std::string_view top_long_short_account_url("topLongShortAccountRatio?");
  static constexpr std::string_view top_long_short_position_url("topLongShortPositionRatio?");
  static constexpr std::string_view global_long_short_account_url("globalLongShortAccountRatio?");
  static constexpr std::string_view taker_buy_sell_url("takerlongshortRatio?");
  static constexpr std::string_view open_interest_hist_url("openInterestHist?");

  std::string url_type;
  std::string_view url;
  if (auto err = obj["url"].get_string().get(url); !err) {
    if (url.rfind(kline_url) != std::string::npos) {
      url_type = "klines_rest";
      const std::string_view interval = "&interval=";
      size_t pos = url.find(interval);
      size_t sep = url.find("&", pos + 1);
      std::string interval_channel =  [&]()->std::string{auto ret = url.substr(pos, sep - pos -1); return {ret.begin(), ret.end()};}();
      if (!CheckChannelFilter("?symbol={}" + interval_channel)) {
        return;
      }
    } else if (url.rfind(wallet_url) != std::string::npos) {
      url_type = "wallet";
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    } else if (url.rfind(top_long_short_account_url) != std::string::npos) {
      url_type = top_long_short_account_url;
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    } else if (url.rfind(top_long_short_position_url) != std::string::npos) {
      url_type = top_long_short_position_url;
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    } else if (url.rfind(global_long_short_account_url) != std::string::npos) {
      url_type = global_long_short_account_url;
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    } else if (url.rfind(taker_buy_sell_url) != std::string::npos) {
      url_type = taker_buy_sell_url;
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    } else if (url.rfind(open_interest_hist_url) != std::string::npos) {
      url_type = open_interest_hist_url;
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    }
  }

  std::string_view stream_name;
  if (auto err = obj["stream"].get_string().get(stream_name); !err) {
    const int at_idx = stream_name.find('@');
    const std::string_view native_symbol = stream_name.substr(0, at_idx);
    const std::string_view ch = stream_name.substr(at_idx + 1);
    const std::string native_symbol_str (native_symbol.begin(), native_symbol.end());

    // parse this msg?
    // e.g. "{}@aggTrade"
    if (!CheckChannelFilter(std::string("{}@") + std::string(ch.begin(), ch.end()))) {
      return;
    }
    simdjson::ondemand::object msg = obj["data"].get_object();

    if (ch == "bookTicker") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (parser_config_.has_sampling_frequency_bps()) {
          if (BinanceBaseFeedUpdateParser::ParseL1RealtimeBookItem(
              book, msg, native_symbol, packet.timestamp)) {
            if (upd_mgr_.BookExchangeTimeCheck()) {
              bool sampled = book->SampleByFrequency(
                  book->Bid0(),
                  book->Ask0(),
                  parser_config_.sampling_frequency_bps());
              if (parser_config_.has_trade_sampling_frequency_bps()) {
                auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                if (sampled ||
                    timespan > parser_config_.trade_sampling_time_length()) {
                  auto* trade = upd_mgr_.GetTradeMutable();
                  if (slim_trade->PopulateTradeAndClear(trade)) {
                    upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
                  }
                }
              }
              if (sampled) {
                upd_mgr_.PublishBook();
              }
            }
          }
        } else {
          if (BinanceBaseFeedUpdateParser::ParseL1RealtimeBookItem(
                  book, msg, native_symbol, packet.timestamp)) {
            upd_mgr_.PublishBook();
          }
        }
      }
    } else if (ch == "aggTrade" || ch == "trade") {
      // please subscribe aggTrade, instead of trade
      const int64_t time_e = msg["E"].get_int64();
      const int64_t time_t = msg["T"].get_int64();

      double tsdiff_sec = (time_e - time_t) * 1e-3;
      constexpr int64_t kTsDiffCutoffSec = 60;
      if (tsdiff_sec > kTsDiffCutoffSec) {
        LOG(ERROR) << "drop due to E - T tsdiff : " << tsdiff_sec;
      } else if (upd_mgr_.is_symbol_registered(native_symbol)) {
        Trade _trade{};
        BinanceBaseFeedUpdateParser::ParseTradeItem(msg, &_trade, time_t);
        if (parser_config_.has_trade_sampling_frequency_bps()) {
          auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
          bool aggregated = slim_trade->TryAggregateSlimTrade(
              packet.timestamp,
              _trade.price,
              _trade.fill_qty,
              _trade.side,
              _trade.timestamp,
              parser_config_.trade_sampling_frequency_bps());
          int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
          if (!aggregated ||
              timespan > parser_config_.trade_sampling_time_length()) {
            auto* trade = upd_mgr_.GetTradeMutable();
            CHECK(slim_trade->PopulateTradeAndClear(trade));
            upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
          }
        } else {
          auto* trade = upd_mgr_.GetTradeMutable();
          *trade = _trade;
          upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
        }
      }
    } else if (ch == "depth@0ms" || ch == "depth@100ms") {  // diff book
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[std::string(native_symbol)];
        if (stat.snapshot_received || packet.timestamp < SNAPSHOT_ADDED_TS) {
          if (publish_bbo_book_only_) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            if (BinanceBaseFeedUpdateParser::ParseBook(
                    book,
                    msg,
                    upd_mgr_.mea().market,
                    BookBuilder::LIGHT_BBO,
                    native_symbol,
                    packet.timestamp)) {
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            }
          } else if (parser_config_.has_sampling_frequency_bps()) {
            if (BinanceBaseFeedUpdateParser::ParseBook(
                    book,
                    msg,
                    upd_mgr_.mea().market,
                    BookBuilder::LIGHT_BBO,
                    native_symbol,
                    packet.timestamp)) {
              if (upd_mgr_.BookExchangeTimeCheck()) {
                bool sampled = book->SampleByFrequency(
                    book->Bid0(),
                    book->Ask0(),
                    parser_config_.sampling_frequency_bps());
                if (parser_config_.has_trade_sampling_frequency_bps()) {
                  auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                  int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                  if (sampled ||
                      timespan > parser_config_.trade_sampling_time_length()) {
                    auto* trade = upd_mgr_.GetTradeMutable();
                    if (slim_trade->PopulateTradeAndClear(trade)) {
                      upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
                    }
                  }
                }
                if (sampled) {
                  upd_mgr_.PublishBook();
                } 
              }
            }
          } else {
            if (BinanceBaseFeedUpdateParser::ParseBook(
                    book,
                    msg,
                    upd_mgr_.mea().market,
                    BookBuilder::DELTA,
                    native_symbol,
                    packet.timestamp)) {
              upd_mgr_.PublishBook();
            }
          }
          stat.update_received = true;
        }
      }
    } else if (ch == "depth20@100ms" || ch == "depth10@100ms" || ch == "depth5@100ms" ||
               ch == "depth5@500ms" || ch == "depth5") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[std::string(native_symbol)];
        if (!stat.snapshot_received || !stat.update_received) {
          if (BinanceBaseFeedUpdateParser::ParseBook(
              book,
              msg,
              upd_mgr_.mea().market,
              BookBuilder::SNAPSHOT,
              native_symbol,
              packet.timestamp)) {
            upd_mgr_.PublishBook();
            stat.snapshot_received = true;
          }
        }
      }
    } else if (ch == "forceOrder") {
      int64_t msg_E = msg["E"].get_int64();
      simdjson::ondemand::object msg_o = msg["o"].get_object();
      int64_t msg_o_T = msg_o["T"].get_int64();

      double tsdiff_sec = (msg_E - msg_o_T) * 1e-3;
      constexpr int64_t kTsDiffCutoffSec = 60;
      if (tsdiff_sec > kTsDiffCutoffSec) {
        LOG(ERROR) << "drop due to E - T tsdiff : " << tsdiff_sec;
      } else if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* liquidation_order = upd_mgr_.GetLiquidationOrderMutable();
        BinanceBaseFeedUpdateParser::ParseLiquidationOrder(msg_o, liquidation_order, msg_o_T);
        upd_mgr_.PublishLiquidationOrderWithExchangeTimeCheck(native_symbol_str, false);
      }
    } else if (ch == "markPrice@1s") {
      const int64_t timestamp = msg["E"].get_int64();
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        if (!funding_rate_only_){
          auto* mark_price = upd_mgr_.GetMarkPriceMutable();
          BinanceBaseFeedUpdateParser::ParseMarkPrice(msg, mark_price, timestamp);
          upd_mgr_.PublishMarkPriceWithExchangeTimeCheck(native_symbol_str);

          if (upd_mgr_.mea() == MarketExchangeApi{Futures, Binance, "v1"}) {
            upd_mgr_.mutable_update()->ResetFeedUpdate();
            auto* index = upd_mgr_.GetIndexMutable();
            BinanceBaseFeedUpdateParser::ParseIndex(msg, index, timestamp);
            upd_mgr_.PublishIndexWithExchangeTimeCheck(native_symbol_str);
          }
        }
        auto norm = upd_mgr_.GetRelativeNorm(native_symbol_str);
        if (norm.find("PERPETUAL") != std::string::npos) {
          upd_mgr_.mutable_update()->ResetFeedUpdate();
          auto* funding_rate = upd_mgr_.GetFundingRateMutable();
          BinanceBaseFeedUpdateParser::ParseFundingRate(msg, funding_rate, timestamp);
          upd_mgr_.PublishFundingRateWithExchangeTimeCheck(native_symbol_str);
        }
      }
    } else if (ch == "indexPrice@1s") {
      auto* index = upd_mgr_.GetIndexMutable();
      std::string underlying_spot;
      const int64_t timestamp = msg["E"].get_int64();
      BinanceBaseFeedUpdateParser::ParseIndex(msg, index, &underlying_spot, timestamp);

      for (const auto& product : upd_mgr_.products()) {
        if (product->native_symbol().find(underlying_spot) != std::string::npos) {
          upd_mgr_.PublishIndexWithExchangeTimeCheck(product->feed_native_symbol());
        }
      }
    } else if (ch == "tokenNav") {
      if (nav_native_to_native_.find(native_symbol_str) == std::end(nav_native_to_native_)) {
        auto product = binance::symbology::BinanceProduct::FromNavNativeProductNoThrow(
            native_symbol_str, packet.timestamp);
        if (!product) {
          return;
        }
        nav_native_to_native_[native_symbol_str] = product->feed_native_symbol();
      }
      const auto& converted_native_symbol = nav_native_to_native_[native_symbol_str];
      if (upd_mgr_.is_symbol_registered(converted_native_symbol)) {
        auto* nav = upd_mgr_.GetNavMutable();
        BinanceBaseFeedUpdateParser::ParseNav(msg, nav);
        upd_mgr_.PublishNavWithExchangeTimeCheck(converted_native_symbol);
      }
    } else if (ch == "ticker") {
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol,
                                             fastfeed::proto::KLINE_INTERVAL_ROLLING_1DAY);
      if (kline != nullptr) {
        kline->SnapshotClear();
        if (BinanceBaseFeedUpdateParser::ParseTicker(msg, kline)) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    } else if (StringStartsWith(ch, "kline_")) {
      std::vector<std::string> splited = Split(ch, "_");
      CHECK_EQ(splited.size(), 2);
      auto interval = BinanceBaseFeedUpdateParser::ParseKlineInterval(splited[1]);
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {
        kline->UpdateClear();
        if (BinanceBaseFeedUpdateParser::ParseKlineWs(msg, kline, interval)) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    } else {
      if (parser_config_.continue_on_error()) {
        LOG(ERROR) << "[Binance processor] unknown stream: " << stream_name;
      } else {
        NOTREACHED() << stream_name;
      }
    }
  } else if (obj["result"].error() != simdjson::NO_SUCH_FIELD) {
    // subscribe result
  } else if (obj["openInterest"].error() != simdjson::NO_SUCH_FIELD) {
    if (!CheckChannelFilter("open_interest")) {
      return;
    }
    // binance stream uses lower characters, but uses uppper in data
    std::string_view native_symbol = obj["symbol"].get_string();
    std::string native_symbol_registered = StringToLowerCopy(std::string(native_symbol.begin(), native_symbol.end()));
    if (upd_mgr_.is_symbol_registered(native_symbol_registered)) {
        auto* open_interest = upd_mgr_.GetOpenInterestMutable();
        BinanceBaseFeedUpdateParser::ParseOpenInterest(obj, open_interest);
        upd_mgr_.PublishOpenInterestWithExchangeTimeCheck(native_symbol_registered);
    }
  } else if (url_type == "klines_rest") {
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);

    std::string native_symbol = parameters["symbol"];
    // futures kline symbol is lower, spot kline symbol is upper,
    // native symbol always is lower
    StringToLowerInPlace(&native_symbol);
    std::string kline_interval = parameters["interval"];
    auto interval = BinanceBaseFeedUpdateParser::ParseKlineInterval(kline_interval);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      simdjson::ondemand::array data = obj["data"].get_array();
      std::optional<int64_t> first_open_timestamp;
      if (data.count_elements() > 0) {
        first_open_timestamp = data.at_pointer("/0/0").get<int64_t>() * 1'000'000LL;
        data.reset();
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (simdjson::ondemand::array kline_data : data) {
        bool freezed = BinanceBaseFeedUpdateParser::ParseKlineRest(
          kline_data,
          klines, 
          interval, 
          packet.timestamp
        );
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else if (url_type == "wallet") {
    simdjson::ondemand::array arr;
    if (!obj["data"].get_array().get(arr)) {
      for (auto status : arr) {
        std::string_view currency = status["coin"].get_string();
        for (const auto& product : upd_mgr_.products()) {
          if (product->base().native_currency() == currency) {
            Wallet* wallet = upd_mgr_.GetWalletMutable();
            BinanceBaseFeedUpdateParser::ParseWallet(status, wallet);
            upd_mgr_.PublishWallet(product->feed_native_symbol());
          }
        }
      }
    }
  } else if (url_type == top_long_short_account_url) {
    auto trading_data = std::bind(&FeedUpdateManager::GetTopLongShortAccountRatioMutable,
        &upd_mgr_, std::placeholders::_1, std::placeholders::_2);
    auto parser = std::bind(&BinanceBaseFeedUpdateParser::ParseAccountRatio<TopLongShortAccountRatio>,
        std::placeholders::_1, std::placeholders::_2);
    auto publisher = std::bind(&FeedUpdateManager::PublishTopLongShortAccountRatio, &upd_mgr_);
    ParseTradingData(obj, url, packet.timestamp, trading_data, parser, publisher);
  } else if (url_type == top_long_short_position_url) {
    auto trading_data = std::bind(&FeedUpdateManager::GetTopLongShortPositionRatioMutable,
        &upd_mgr_, std::placeholders::_1, std::placeholders::_2);
    auto parser = std::bind(&BinanceBaseFeedUpdateParser::ParsePositionRatio,
        std::placeholders::_1, std::placeholders::_2);
    auto publisher = std::bind(&FeedUpdateManager::PublishTopLongShortPositionRatio, &upd_mgr_);
    ParseTradingData(obj, url, packet.timestamp, trading_data, parser, publisher);
  } else if (url_type == global_long_short_account_url) {
    auto trading_data = std::bind(&FeedUpdateManager::GetGlobalLongShortAccountRatioMutable,
        &upd_mgr_, std::placeholders::_1, std::placeholders::_2);
    auto parser = std::bind(&BinanceBaseFeedUpdateParser::ParseAccountRatio<GlobalLongShortAccountRatio>,
        std::placeholders::_1, std::placeholders::_2);
    auto publisher = std::bind(&FeedUpdateManager::PublishGlobalLongShortAccountRatio, &upd_mgr_);
    ParseTradingData(obj, url, packet.timestamp, trading_data, parser, publisher);
  } else if (url_type == taker_buy_sell_url) {
    auto trading_data = std::bind(&FeedUpdateManager::GetTakerBuySellRatioMutable,
        &upd_mgr_, std::placeholders::_1, std::placeholders::_2);
    auto parser = std::bind(&BinanceBaseFeedUpdateParser::ParseTakerRatio,
        std::placeholders::_1, std::placeholders::_2);
    auto publisher = std::bind(&FeedUpdateManager::PublishTakerBuySellRatio, &upd_mgr_);
    ParseTradingData(obj, url, packet.timestamp, trading_data, parser, publisher);
  } else if (url_type == open_interest_hist_url) {
    auto trading_data = std::bind(&FeedUpdateManager::GetOpenInterestHistMutable,
        &upd_mgr_, std::placeholders::_1, std::placeholders::_2);
    auto parser = std::bind(&BinanceBaseFeedUpdateParser::ParseOpenInterestHist,
        std::placeholders::_1, std::placeholders::_2);
    auto publisher = std::bind(&FeedUpdateManager::PublishOpenInterestHist, &upd_mgr_);
    ParseTradingData(obj, url, packet.timestamp, trading_data, parser, publisher);
  } else {
    if (parser_config_.continue_on_error()) {
      LOG(ERROR) << "[BinanceCommonProcessor] unknown raw msg: " << packet.payload;
    } else {
      NOTREACHED() << "[BinanceCommonProcessor] unknown raw msg: " << packet.payload;
    }
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::binance_base::feed_v1::impl

}  // namespace coin2::exchange::binance_base::feed_v1::impl
